package cn.azzhu.day05

import org.apache.flink.api.common.functions.JoinFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{TumblingEventTimeWindows, TumblingTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * 基于窗口的join
 *
 * @author azzhu
 * @create 2020-09-21 20:33:28
 */
object WindowJoinExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //基于间隔的join，只能使用事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val input1 = env.fromElements(
      ("a",1,1000L),
      ("a",2,2000L),
      ("b",1,3000L),
      ("b",2,4000L)
    )
      .assignAscendingTimestamps(_._3)

    val input2 = env.fromElements(
      ("a",10,1000L),
      ("a",20,2000L),
      ("b",10,3000L),
      ("b",20,4000L)
    )
      .assignAscendingTimestamps(_._3)

   input1.join(input2)
     // where equalTo ====> on input1._1 = input2_1
     .where(_._1)
     .equalTo(_._1)
     .window(TumblingEventTimeWindows.of(Time.seconds(10)))
     .apply(new MyJoin)
     .print()

    env.execute("WindowJoinExample")
  }

  //分流开窗口以后，属于同一个窗口的input1中的元素和inpu2中的元素做笛卡尔积
  //相同的key，且是相同的窗口中的元素做笛卡尔积
  class MyJoin extends JoinFunction[(String,Int,Long),(String,Int,Long),String] {
    override def join(in1: (String, Int, Long), in2: (String, Int, Long)): String = {
      in1 + "====>" + in2
    }
  }
}
